package com.py.gs.util.base.concurrent;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.py.gs.tools.JBuilder;
import com.py.gs.util.net.util.Msg;

/**
 * 
 * @author KXZ 
 * 2018年6月26日
 */
public class FileCache extends Thread {
	
	private static Logger log = Logger.getLogger(FileCache.class);
	static final byte[] RESULT_ALL_HEAD = "{data:[{SN:".getBytes();
	static final byte[] RESULT_HEAD = "{SN:".getBytes();
	static final byte[] RESULT_TYPE = ",type:".getBytes();
	static final byte[] RESULT_DATA = ",data:".getBytes();
	static final byte[] RESULT_END = "},".getBytes();
	static final int LENGTH_INIT = RESULT_ALL_HEAD.length;
	static final int LENGTH_PACKAGE = RESULT_HEAD.length+RESULT_TYPE.length+RESULT_DATA.length+RESULT_END.length+19+5;
	
	FileCacheReadHandle readHandle;
	
	RandomAccessFile fileWriting = null;
	RandomAccessFile fileReading = null;
	RandomAccessFile fileIdx = null;
	/** 读线程是否已经要开始消费写线程的数据 */
	boolean beginReadCount = false;
	SingleCounter unreadCount = new SingleCounter(0);
	String dataPath;
	/** 标记上一次写索引 */
	int idxLastWriteFile = 0;
	int idxLastWritePos = 0;
	int idxWriteFile = 0;
	int idxReadFile = 0;
	int idxReadPos = 0;
	byte[] dataIdx = new byte[8];
	byte[] dataNextFile = new byte[4];
	byte[] readCache = new byte[1024*1024];//32768
	int readCachePos = 0;
	byte[] resultCache = new byte[readCache.length+readCache.length/2];
	BytesArray resultBytesArray = new BytesArray(resultCache);
	
	int resultPos = 0;
	int readPos = 0;
	
	int maxFileSize = 1024*1024*100;
	CountDownLatch readLock = new CountDownLatch(1);
	volatile long lastSN = 0;
	volatile long beginSN = 0;
	
	public ArrayList<Object[]> caches = new ArrayList<>(16);
	public int cacheSize = 0;
	ThreadPoolExecutor cacheExecutor;

	
	/** 该文件缓存的路径, 必须以/结尾 */
	public FileCache(String dataPath, FileCacheReadHandle readHandle) {
		this.dataPath = dataPath;
		this.readHandle = readHandle;
		this.setName("cache-"+dataPath);
		cacheExecutor = new ThreadPoolExecutor(1, 1, 24L, TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("fileCacheWrite"+dataPath));
		appendMsg(RESULT_ALL_HEAD);
		try {
			File path = new File(dataPath);
			if(!path.exists()) {
				path.mkdirs();
			}
			File f = new File(dataPath+"idx.txt");
			if(!f.exists()){
				f.createNewFile();
			}
			fileIdx = new RandomAccessFile(f, "rw");
			if(fileIdx.length() >= 8) {
				fileIdx.read(dataIdx);
				idxReadFile = Msg.readInt(dataIdx, 0);
				idxReadPos = Msg.readInt(dataIdx, 4);
			}
			f = new File(dataPath+idxReadFile+".txt");
			if(!f.exists()) {
				setReadIdx(0, 0);
				for (File file : path.listFiles()) {
					if(!file.getName().equals("idx.txt")) {
						file.delete();
					}
				}
			}else {//找出最大的写索引
				idxLastWriteFile = idxReadFile;
				for (File file : path.listFiles()) {
					String fn = file.getName();
					if(!fn.equals("idx.txt")) {
						int idx = Integer.parseInt(fn.substring(0, fn.length()-4));
						if(idx >= idxLastWriteFile) {
							idxLastWriteFile = idx;
							idxLastWritePos = (int) file.length();
						}
					}
				}
			}
			fileWriting = new RandomAccessFile(dataPath+idxLastWriteFile+".txt", "rw");
			fileWriting.seek(idxLastWritePos);
			fileReading = new RandomAccessFile(dataPath+idxReadFile+".txt", "rw");
			fileReading.seek(idxReadPos);
			idxWriteFile = idxLastWriteFile;
			checkBeginReadCount();
			//BaseGame.get().addFileCache(this, path);
		} catch (Exception e) {
			log.error(dataPath+"文件缓存初始化报错", e);
			System.exit(0);
		}
		
	}
	
	private void appendMsg(byte[] resultHead) {
		for (int i = 0; i < resultHead.length; i++) {
			resultCache[i+resultPos] = resultHead[i];
		}
		resultPos+=resultHead.length;
	}

	private void checkBeginReadCount() {
		if(idxReadFile < idxLastWriteFile || (idxReadFile == idxLastWriteFile && idxReadPos < idxLastWritePos)){
			return;
		}
		beginReadCount = true;
	}
	
	/** 设置读索引并写入文件 */
	private void setReadIdx(int idxFile, int idxPos) {
		this.idxReadFile = idxFile;
		this.idxReadPos = idxPos;
		try {
			fileIdx.seek(0);
			Msg.writeInt(dataIdx, 0, idxFile);
			Msg.writeInt(dataIdx, 4, idxPos);
			fileIdx.write(dataIdx);
		} catch (IOException e) {
			log.error(dataPath+"写入索引文件失败", e);
		}
	}
	
	/** 写入json数据, 写入的消息不能拿去复用
	 * @param 消息类型 0-65535
	 */
	public void write(int type, Object o) {
		caches.add(new Object[]{type, o});
	}
	
	public Future<?> cacheCommit(boolean flush, boolean getFuture){
		if(caches.size()==0){
			return null;
		}
		if(!flush){
			cacheSize = caches.size();
			return null;
		}
		
		final ArrayList<Object[]> cs = caches;
		caches = new ArrayList<>();
		cacheSize = 0;
		if (getFuture) {
			return cacheExecutor.submit(()->flush(cs));
		} else {
			cacheExecutor.execute(()->flush(cs));
			return null;
		}
		
//		flush(caches);
//		caches.clear();
//		cacheSize = 0;
	}

	public void cacheCancel() {
		for (int i = caches.size()-1; i >= cacheSize; i--) {
			caches.remove(i);
		}
	}
	
	/** 最多缓存4M, 实际最多是100ms生成 */
	JBuilder builder = new JBuilder(4000000);
	long lastStartTime = 0;
	int totalSize = 0;
	public void flush(ArrayList<Object[]> cs) {
		int size = cs.size();
		try {
			if(builder.count == 0){
				lastStartTime = System.currentTimeMillis();
			}
			for (int i = 0; i < size; i++) {
				Object[] msg = cs.get(i);
				Object msg1 = msg[1];
				if(msg1 instanceof BigJsonData){
					builder.begin((Integer)msg[0]);
					((BigJsonData)msg[1]).toJson(builder);
					builder.end();
				}else{
					byte[] bs = JSON.toJSONBytes(msg[1]);
					builder.addJson((Integer)msg[0], bs);
				}
			}
			totalSize+=size;
			if(cacheExecutor.getQueue().size()==0 || builder.count > 3500000 || System.currentTimeMillis()-lastStartTime>100){
//				System.out.println("写入"+builder.count+","+(System.currentTimeMillis()-lastStartTime));
				fileWriting.write(builder.buf, 0, builder.count);
				unreadCount.countDown(totalSize);
				totalSize = 0;
				builder.setLength(0);
				checkNextWriteFile();
//				lastStartTime = 0;
			}
//			fileWriting.write(builder.buf, 0, builder.count);
//			unreadCount.countDown(size);
//			builder.setLength(0);
//			checkNextWriteFile();
		} catch (IOException e) {
			log.error(dataPath+"写入数据文件失败", e);
		}
	}


	private void checkNextWriteFile() throws IOException {
		if(fileWriting.length()>maxFileSize) {
			fileWriting.write(dataNextFile);
			idxWriteFile++;
			fileWriting.close();
			fileWriting = new RandomAccessFile(dataPath+idxWriteFile+".txt", "rw");
		}
	}
	
	static class Test {
		public String a = "a";
		public int i = 1;
	}

	static CountDownLatch testLock = new CountDownLatch(1);
	public static void main(String[] args) {
		FileCache fc = new FileCache("testFileCache/bigData/", null);
		fc.start();
		fc.readLock.countDown();
		
		new Thread(()->{//大数据量写数据线程
			for (int i = 0; i < 1000000; i++) {
				fc.write(1, new Test());
				fc.cacheCommit(true, false);
			}
		}).start();
		

		
		new Thread(()->{//10s一次的写数据线程
			for (int i = 0; i < 100000; i++) {
				fc.write(1, new Test());
				fc.cacheCommit(true, false);
				try {
					Thread.sleep(3000);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
		
		new Thread(()->{//保存已读游标线程
			for (;;) {
				try {
					Thread.sleep(100);//模拟 发送完数据之后20ms返回
					testLock.await();
					testLock.countAdd(1);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				System.out.println("saveReadPos:" + fc.lastSN);
				fc.saveReadPos(fc.lastSN);
			}
		}).start();
	}

	@Override
	public void run() {
		int msgCount = 0;
		boolean toReadNextFile = false;
		mainLoop:while(true) {
			try {
				System.out.println("msgCount-before:" + msgCount);
//				System.out.println("run------0");
				if(beginReadCount) {//启动时文件已经读完，并且上一次的消息都处理结束才等待
					if (msgCount <= 0) {
						msgCount += unreadCount.await();
					} else {
						int newMsgCount = (int)-unreadCount.getCount();
						if (newMsgCount > 0) {
							unreadCount.countAdd(newMsgCount);
							msgCount += newMsgCount;
						}
					}
					System.out.println("msgCount-wait.....");
				}
				System.out.println("msgCount-end:" + msgCount);
//				System.out.println("run------1");
				if(toReadNextFile){
					idxReadFile ++;
					readNextFile();
					idxReadPos = 0;
					readCachePos = 0;
					toReadNextFile = false;
				}
				int readBytesCnt = Math.max(0, fileReading.read(readCache, readPos, readCache.length-readPos))+readPos;
				if(readBytesCnt==-1) {
					log.error(dataPath+"文件没读到消息,"+idxReadFile+"."+fileReading.getFilePointer());
					continue;
				}
				readCachePos = 0;
				int len=0;
				long SN = 0;
				for (; readCachePos < readBytesCnt; ) {
					if(readBytesCnt - readCachePos < 4) {
						//被截断的内容
						System.arraycopy(readCache, readCachePos, readCache, 0, readBytesCnt - readCachePos);
						readPos = readBytesCnt - readCachePos;
						break;
					}
					len = Msg.readInt(readCache, readCachePos);
					if(len == 0) {//开始读取下一个文件
						toReadNextFile = true;
						break;
					}
					if(len+4+readCachePos > readCache.length) {
						if(readCachePos==0) {//至少2倍扩容
							byte[] newCache = new byte[Math.max(len+4, readCache.length*2)];
							System.arraycopy(readCache, 0, newCache, 0, readCache.length);
							readPos = readCache.length;
							readCache = newCache;
							continue mainLoop;
						}
						//被截断的内容
						System.arraycopy(readCache, readCachePos, readCache, 0, readBytesCnt - readCachePos);
						readPos = readBytesCnt - readCachePos;
						break;
					}
					SN = appendMsg(len);
					len += 4;
					idxReadPos += len;
					readCachePos += len;
					if(beginReadCount) {
						if(--msgCount==0) {
							break;
						}
					}else {//直接读到超过上次标记写索引
						checkBeginReadCount();
					}
				}
				if(resultPos == LENGTH_INIT) {//只切换了文件
					continue;
				}
				if(readCachePos == readBytesCnt) {//刚好读完
					readPos = 0;
				} else if(readCachePos > 0){
					System.arraycopy(readCache, readCachePos, readCache, 0, readBytesCnt - readCachePos);
					readPos = readBytesCnt - readCachePos;
				}
				resultCache[resultPos-1] = ']';
				resultCache[resultPos++] = '}';
				
				resultBytesArray.len = resultPos;
				try {//new String(msgs, "utf8")
					lastSN = SN;
					readLock.await();//等待管理后台连接成功
					while(SN == lastSN) {
						readLock.countAdd(1);
						//readHandle.onRead(resultBytesArray);
						testLock.countDown();
						readLock.await();//等待管理后台响应
					}
					beginSN = SN;
				} catch (Throwable e) {
					log.error(dataPath+"onRead处理消息出错:"+idxReadFile, e);
				}
				resultPos = LENGTH_INIT;
			} catch (Exception e) {
				log.error(dataPath+"读文件出错"+idxReadFile, e);
			}
		}
		
	}
	
	private long appendMsg(int len) {
		len = len-2;
		if(resultPos>LENGTH_INIT) {
			if(resultPos + LENGTH_PACKAGE + len > resultCache.length) {
				byte[] cache =  new byte[Math.max(resultPos + LENGTH_PACKAGE + len, resultCache.length*2)];
				System.arraycopy(resultCache, 0, cache, 0, resultPos);
				resultCache = cache;
				resultBytesArray.data = cache;
			}
			appendMsg(RESULT_HEAD);
		}
		int type = Msg.readShort(readCache, readCachePos+4);
		long sn = ((idxReadFile*1L)<<32) | (idxReadPos+len+6);
		resultPos += Msg.writeLongStr(resultCache, resultPos, sn);
		appendMsg(RESULT_TYPE);
		resultPos += Msg.writeNumStr(resultCache, resultPos, type);
		appendMsg(RESULT_DATA);
		System.arraycopy(readCache, readCachePos+6, resultCache, resultPos, len);
		resultPos+=len;//new String(resultCache, 0, resultPos, "utf-8");new String(readCache, readCachePos+6, len, "utf-8");
		appendMsg(RESULT_END);
		return sn;
	}

	private void readNextFile() {
		readPos = 0;
		try {
			fileReading.close();
			fileReading = new RandomAccessFile(dataPath+idxReadFile+".txt", "rw");
		} catch (Exception e) {
			log.error(dataPath+"读下一个数据文件出错"+idxReadFile, e);
		}
	}
	
	/** 得到回复保存最新读游标 */
	public void saveReadPos(long SN) {
		if(SN == lastSN) {
		}else{//接收方没接收成功暂时不管
			log.error(dataPath+"读取文件SN改变:"+SN+", 原本:"+beginSN+"/"+lastSN);
			return;
		}
		int idxFile = (int)(SN>>32);
		int idxPos = (int)(SN&0xffffffff);
		if(idxFile != idxReadFile) {//需要改变读取文件
			log.warn("读取文件索引改变:"+idxFile+", 原本:"+idxReadFile);
			idxReadFile = idxFile;
			readNextFile();
		}
		if(idxPos != idxReadPos) {
			try {
				log.info("读取文件游标改变:"+idxPos+", 原本:"+idxReadPos);
				if(idxPos>fileReading.length()) {
					throw new RuntimeException("读取文件游标大于文件长度,"+idxReadFile+"."+idxPos);
				}
				idxReadPos = idxPos;
				fileReading.seek(idxReadPos);
				readPos = 0;
			} catch (IOException e) {
				log.error("读取文件失败:"+idxFile+"."+idxPos, e);
			}
		}
		setReadIdx(idxFile, idxPos);
		lastSN = 0;//重置跳出循环
		readLock.countDown();
	}

	public FileCacheReadHandle getReadHandle() {
		return readHandle;
	}

	public void setReadHandle(FileCacheReadHandle readHandle) {
		this.readHandle = readHandle;
	}

	/** 网关重连, 或者由网关通知重推时
	 * @param reconnect 是否重连
	 */
	public void continueRead(boolean reconnect) {
		readLock.countDown();
	}
	
	
	
}
